{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "fa505c58",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| default_exp core"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2161d9b5",
   "metadata": {},
   "source": [
    "# Core\n",
    "> The `FastHTML` subclass of `Starlette`, along with the `RouterX` and `RouteX` classes it automatically uses."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "46e2e6e8",
   "metadata": {},
   "source": [
    "This is the source code to fasthtml. You won't need to read this unless you want to understand how things are built behind the scenes, or need full details of a particular API. The notebook is converted to the Python module [fasthtml/core.py](https://github.com/AnswerDotAI/fasthtml/blob/main/fasthtml/core.py) using [nbdev](https://nbdev.fast.ai/)."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "568b20cd",
   "metadata": {},
   "source": [
    "## Imports and utils"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "23503b9e",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "import json,uuid,inspect,types,signal,asyncio,threading,inspect,random,contextlib,httpx,itsdangerous\n",
    "\n",
    "from fastcore.utils import *\n",
    "from fastcore.xml import *\n",
    "from fastcore.meta import use_kwargs_dict\n",
    "\n",
    "from types import UnionType, SimpleNamespace as ns, GenericAlias\n",
    "from typing import Optional, get_type_hints, get_args, get_origin, Union, Mapping, TypedDict, List, Any\n",
    "from datetime import datetime,date\n",
    "from dataclasses import dataclass,fields\n",
    "from collections import namedtuple\n",
    "from inspect import isfunction,ismethod,Parameter,get_annotations\n",
    "from functools import wraps, partialmethod, update_wrapper\n",
    "from http import cookies\n",
    "from urllib.parse import urlencode, parse_qs, quote, unquote\n",
    "from copy import copy,deepcopy\n",
    "from warnings import warn\n",
    "from dateutil import parser as dtparse\n",
    "from anyio import from_thread\n",
    "from uuid import uuid4, UUID\n",
    "from base64 import b85encode,b64encode\n",
    "\n",
    "from fasthtml.starlette import *"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "7f5d0a72",
   "metadata": {},
   "outputs": [],
   "source": [
    "import time\n",
    "\n",
    "from IPython import display\n",
    "from enum import Enum\n",
    "from pprint import pprint\n",
    "\n",
    "from fastcore.test import *\n",
    "from starlette.testclient import TestClient\n",
    "from starlette.requests import Headers\n",
    "from starlette.datastructures import UploadFile"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "19d3f2a7",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def _params(f): return signature_ex(f, True).parameters\n",
    "\n",
    "empty = Parameter.empty"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "19ffa7e0",
   "metadata": {},
   "source": [
    "We write source code _first_, and then tests come _after_. The tests serve as both a means to confirm that the code works and also serves as working examples. The first exported function, `parsed_date`, is an example of this pattern."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "e68a76c9",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def parsed_date(s:str):\n",
    "    \"Convert `s` to a datetime\"\n",
    "    return dtparse.parse(s)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "5331a3e7",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "datetime.datetime(2025, 11, 19, 14, 0)"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "parsed_date('2pm')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "c40c9071",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "True"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "isinstance(date.fromtimestamp(0), date)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "7c820373",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def snake2hyphens(s:str):\n",
    "    \"Convert `s` from snake case to hyphenated and capitalised\"\n",
    "    s = snake2camel(s)\n",
    "    return camel2words(s, '-')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "442a5aac",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'Snake-Case'"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "snake2hyphens(\"snake_case\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "25f3b8f8",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "htmx_hdrs = dict(\n",
    "    boosted=\"HX-Boosted\",\n",
    "    current_url=\"HX-Current-URL\",\n",
    "    history_restore_request=\"HX-History-Restore-Request\",\n",
    "    prompt=\"HX-Prompt\",\n",
    "    request=\"HX-Request\",\n",
    "    target=\"HX-Target\",\n",
    "    trigger_name=\"HX-Trigger-Name\",\n",
    "    trigger=\"HX-Trigger\")\n",
    "\n",
    "@dataclass\n",
    "class HtmxHeaders:\n",
    "    boosted:str|None=None; current_url:str|None=None; history_restore_request:str|None=None; prompt:str|None=None\n",
    "    request:str|None=None; target:str|None=None; trigger_name:str|None=None; trigger:str|None=None\n",
    "    def __bool__(self): return any(hasattr(self,o) for o in htmx_hdrs)\n",
    "\n",
    "def _get_htmx(h):\n",
    "    res = {k:h.get(v.lower(), None) for k,v in htmx_hdrs.items()}\n",
    "    return HtmxHeaders(**res)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "5b4b5d95",
   "metadata": {},
   "outputs": [],
   "source": [
    "def test_request(url: str='/', headers: dict=None, method: str='get')->Request:\n",
    "    if headers is None: headers={}\n",
    "    scope = {\n",
    "        'type': 'http', 'method': method, 'path': url,\n",
    "        'headers': Headers(headers).raw,\n",
    "        'query_string': b'', 'scheme': 'http',\n",
    "        'client': ('127.0.0.1', 8000),\n",
    "        'server': ('127.0.0.1', 8000),\n",
    "    }\n",
    "    receive = lambda: {\"body\": b\"\", \"more_body\": False}\n",
    "    return Request(scope, receive)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "36e2cac0",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "HtmxHeaders(boosted=None, current_url=None, history_restore_request=None, prompt=None, request='1', target=None, trigger_name=None, trigger=None)"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "h = test_request(headers=Headers({'HX-Request':'1'}))\n",
    "_get_htmx(h.headers)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "1d53e8e7",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def _mk_list(t, v): return [t(o) for o in listify(v)]"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ef18f76b",
   "metadata": {},
   "source": [
    "## Request and response"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "5ab74473",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "fh_cfg = AttrDict(indent=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "0afb520c",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def _fix_anno(t, o):\n",
    "    \"Create appropriate callable type for casting a `str` to type `t` (or first type in `t` if union)\"\n",
    "    origin = get_origin(t)\n",
    "    if origin is Union or origin is UnionType or origin in (list,List):\n",
    "        t = first(o for o in get_args(t) if o!=type(None))\n",
    "    d = {bool: str2bool, int: str2int, date: str2date, UploadFile: noop}\n",
    "    res = d.get(t, t)\n",
    "    if origin in (list,List): return _mk_list(res, o)\n",
    "    if not isinstance(o, (str,list,tuple)): return o\n",
    "    return res(o[-1]) if isinstance(o,(list,tuple)) else res(o)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "95b1f5c9",
   "metadata": {},
   "outputs": [],
   "source": [
    "test_eq(_fix_anno(Union[str,None], 'a'), 'a')\n",
    "test_eq(_fix_anno(float, 0.9), 0.9)\n",
    "test_eq(_fix_anno(int, '1'), 1)\n",
    "test_eq(_fix_anno(int, ['1','2']), 2)\n",
    "test_eq(_fix_anno(list[int], ['1','2']), [1,2])\n",
    "test_eq(_fix_anno(list[int], '1'), [1])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "c58ccadb",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def _form_arg(k, v, d):\n",
    "    \"Get type by accessing key `k` from `d`, and use to cast `v`\"\n",
    "    if v is None: return\n",
    "    if not isinstance(v, (str,list,tuple)): return v\n",
    "    # This is the type we want to cast `v` to\n",
    "    anno = d.get(k, None)\n",
    "    if not anno: return v\n",
    "    return _fix_anno(anno, v)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "59757d76",
   "metadata": {},
   "outputs": [],
   "source": [
    "d = dict(k=int, l=List[int])\n",
    "test_eq(_form_arg('k', \"1\", d), 1)\n",
    "test_eq(_form_arg('l', \"1\", d), [1])\n",
    "test_eq(_form_arg('l', [\"1\",\"2\"], d), [1,2])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "5fc04751",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "@dataclass\n",
    "class HttpHeader: k:str;v:str"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "id": "94e18161",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def _to_htmx_header(s): return 'HX-' + s.replace('_', '-').title()\n",
    "\n",
    "htmx_resps = dict(location=None, push_url=None, redirect=None, refresh=None, replace_url=None,\n",
    "                 reswap=None, retarget=None, reselect=None, trigger=None, trigger_after_settle=None, trigger_after_swap=None)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "id": "592d6c8e",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'HX-Trigger-After-Settle'"
      ]
     },
     "execution_count": 21,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "_to_htmx_header('trigger_after_settle')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "id": "f6a2e62e",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "@use_kwargs_dict(**htmx_resps)\n",
    "def HtmxResponseHeaders(**kwargs):\n",
    "    \"HTMX response headers\"\n",
    "    res = tuple(HttpHeader(_to_htmx_header(k), v) for k,v in kwargs.items())\n",
    "    return res[0] if len(res)==1 else res"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "id": "e89857eb",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "HttpHeader(k='HX-Trigger-After-Settle', v='hi')"
      ]
     },
     "execution_count": 23,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "HtmxResponseHeaders(trigger_after_settle='hi')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "id": "56cc589f",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def _annotations(anno):\n",
    "    \"Same as `get_annotations`, but also works on namedtuples\"\n",
    "    if is_namedtuple(anno): return {o:str for o in anno._fields}\n",
    "    return get_annotations(anno)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "id": "5453a684",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'a': int, 'b': str}"
      ]
     },
     "execution_count": 25,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "@dataclass\n",
    "class Foo: a:int; b:str\n",
    "_annotations(Foo)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "id": "bcad3a5a",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'x': str, 'y': str}"
      ]
     },
     "execution_count": 26,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "Point = namedtuple('Point', ['x', 'y'])\n",
    "_annotations(Point)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "id": "cb4ed4aa",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def _is_body(anno): return issubclass(anno, (dict,ns)) or hasattr(anno,'__from_request__') or _annotations(anno)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "id": "ffdde66f",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def _formitem(form, k):\n",
    "    \"Return single item `k` from `form` if len 1, otherwise return list\"\n",
    "    if isinstance(form, dict): return form.get(k)\n",
    "    o = form.getlist(k)\n",
    "    return o[0] if len(o) == 1 else o if o else None"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "id": "5fe74444",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def form2dict(form: FormData) -> dict:\n",
    "    \"Convert starlette form data to a dict\"\n",
    "    if isinstance(form, dict): return form\n",
    "    return {k: _formitem(form, k) for k in form}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "id": "cf80ea34",
   "metadata": {},
   "outputs": [],
   "source": [
    "d = [('a',1),('a',2),('b',0)]\n",
    "fd = FormData(d)\n",
    "res = form2dict(fd)\n",
    "test_eq(res['a'], [1,2])\n",
    "test_eq(res['b'], 0)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "id": "42c9cea0",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "async def parse_form(req: Request) -> FormData:\n",
    "    \"Starlette errors on empty multipart forms, so this checks for that situation\"\n",
    "    ctype = req.headers.get(\"Content-Type\", \"\")\n",
    "    if ctype=='application/json': return await req.json()\n",
    "    if not ctype.startswith(\"multipart/form-data\"): return await req.form()\n",
    "    try: boundary = ctype.split(\"boundary=\")[1].strip()\n",
    "    except IndexError: raise HTTPException(400, \"Invalid form-data: no boundary\")\n",
    "    min_len = len(boundary) + 6\n",
    "    clen = int(req.headers.get(\"Content-Length\", \"0\"))\n",
    "    if clen <= min_len: return FormData()\n",
    "    return await req.form()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "id": "33d3bc16",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "async def _from_body(req, p):\n",
    "    \"Parse request form/query data and create an instance of the annotated parameter type\"\n",
    "    anno = p.annotation\n",
    "    data = form2dict(await parse_form(req))\n",
    "    if req.query_params: data = {**data, **dict(req.query_params)}\n",
    "    ctor = getattr(anno, '__from_request__', None)\n",
    "    if ctor: return await ctor(data, req) if asyncio.iscoroutinefunction(ctor) else ctor(data, req)\n",
    "    d = _annotations(anno)\n",
    "    cargs = {k: _form_arg(k, v, d) for k, v in data.items() if not d or k in d}\n",
    "    return anno(**cargs)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "437263a3",
   "metadata": {},
   "source": [
    "Use the special `__from_request__` class method for custom creation:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "id": "8235f17f",
   "metadata": {},
   "outputs": [],
   "source": [
    "class Foo(BasicRepr):\n",
    "    def __init__(self, d): self.d = d\n",
    "\n",
    "    @classmethod\n",
    "    def __from_request__(cls, data, req): return cls(data)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "id": "a4873835",
   "metadata": {},
   "outputs": [],
   "source": [
    "p = Parameter('a', Parameter.POSITIONAL_OR_KEYWORD, annotation=Foo)\n",
    "\n",
    "req = test_request(method='POST')\n",
    "async def mock_form(): return FormData({'a':1})\n",
    "req.form = mock_form"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "id": "4b36f60b",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Foo(d={'a': 1})"
      ]
     },
     "execution_count": 35,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "await _from_body(req, p)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "id": "7cc39ba9",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "class JSONResponse(JSONResponseOrig):\n",
    "    \"Same as starlette's version, but auto-stringifies non serializable types\"\n",
    "    def render(self, content: Any) -> bytes:\n",
    "        res = json.dumps(content, ensure_ascii=False, allow_nan=False, indent=None, separators=(\",\", \":\"), default=str)\n",
    "        return res.encode(\"utf-8\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "id": "2a8b10f4",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{'k': 'value1', 'v': 'value3'}\n"
     ]
    }
   ],
   "source": [
    "async def f(req):\n",
    "    def _f(p:HttpHeader): ...\n",
    "    p = first(_params(_f).values())\n",
    "    result = await _from_body(req, p)\n",
    "    return JSONResponse(result.__dict__)\n",
    "\n",
    "client = TestClient(Starlette(routes=[Route('/', f, methods=['POST'])]))\n",
    "\n",
    "d = dict(k='value1',v=['value2','value3'])\n",
    "response = client.post('/', data=d)\n",
    "print(response.json())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "id": "9246153f",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "\"['1', '2']\""
      ]
     },
     "execution_count": 38,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "async def f(req): return Response(str(req.query_params.getlist('x')))\n",
    "client = TestClient(Starlette(routes=[Route('/', f, methods=['GET'])]))\n",
    "client.get('/?x=1&x=2').text"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "id": "6775cbf8",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "async def _find_p(req, arg:str, p:Parameter):\n",
    "    \"In `req` find param named `arg` of type in `p` (`arg` is ignored for body types)\"\n",
    "    anno = p.annotation\n",
    "    # If there's an annotation of special types, return object of that type\n",
    "    # GenericAlias is a type of typing for iterators like list[int] that is not a class\n",
    "    if isinstance(anno, type) and not isinstance(anno, GenericAlias):\n",
    "        if issubclass(anno, Request): return req\n",
    "        if issubclass(anno, HtmxHeaders): return _get_htmx(req.headers)\n",
    "        if issubclass(anno, Starlette): return req.scope['app']\n",
    "        if _is_body(anno) and 'session'.startswith(arg.lower()): return req.scope.get('session', {})\n",
    "        if _is_body(anno): return await _from_body(req, p)\n",
    "    # If there's no annotation, check for special names\n",
    "    if anno is empty:\n",
    "        if 'request'.startswith(arg.lower()): return req\n",
    "        if 'session'.startswith(arg.lower()): return req.scope.get('session', {})\n",
    "        if arg.lower()=='scope': return dict2obj(req.scope)\n",
    "        if arg.lower()=='auth': return req.scope.get('auth', None)\n",
    "        if arg.lower()=='htmx': return _get_htmx(req.headers)\n",
    "        if arg.lower()=='app': return req.scope['app']\n",
    "        if arg.lower()=='body': return (await req.body()).decode()\n",
    "        if arg.lower() in ('hdrs','ftrs','bodykw','htmlkw'): return getattr(req, arg.lower())\n",
    "        if arg!='resp': warn(f\"`{arg} has no type annotation and is not a recognised special name, so is ignored.\")\n",
    "        return None\n",
    "    # Look through path, cookies, headers, query, and body in that order\n",
    "    res = req.path_params.get(arg, None)\n",
    "    if res in (empty,None): res = req.cookies.get(arg, None)\n",
    "    if res in (empty,None): res = req.headers.get(snake2hyphens(arg), None)\n",
    "    if res in (empty,None): res = req.query_params.getlist(arg)\n",
    "    if res==[]: res = None\n",
    "    if res in (empty,None): res = _formitem(await parse_form(req), arg)\n",
    "    # Raise 400 error if the param does not include a default\n",
    "    if (res in (empty,None)) and p.default is empty: raise HTTPException(400, f\"Missing required field: {arg}\")\n",
    "    # If we have a default, return that if we have no value\n",
    "    if res in (empty,None): res = p.default\n",
    "    # We can cast str and list[str] to types; otherwise just return what we have\n",
    "    if anno is empty: return res\n",
    "    try: return _fix_anno(anno, res)\n",
    "    except ValueError: raise HTTPException(404, req.url.path) from None\n",
    "\n",
    "async def _wrap_req(req, params):\n",
    "    return {arg:await _find_p(req, arg, p) for arg,p in params.items()}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "id": "bf945ee8",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{'req': <starlette.requests.Request object at 0x11124aa80>, 'this': <starlette.applications.Starlette object at 0x1111e16a0>, 'a': '1', 'b': HttpHeader(k='value1', v='value3')}\n"
     ]
    }
   ],
   "source": [
    "def g(req, this:Starlette, a:str, b:HttpHeader): ...\n",
    "\n",
    "async def f(req):\n",
    "    a = await _wrap_req(req, _params(g))\n",
    "    return Response(str(a))\n",
    "\n",
    "client = TestClient(Starlette(routes=[Route('/', f, methods=['POST'])]))\n",
    "response = client.post('/?a=1', data=d)\n",
    "print(response.text)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "id": "a3ded5ec",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{'req': <starlette.requests.Request object at 0x11124a420>, 'this': <starlette.applications.Starlette object at 0x1111e0a10>, 'a': '1', 'b': HttpHeader(k='value1', v='value3')}\n"
     ]
    }
   ],
   "source": [
    "def g(req, this:Starlette, a:str, b:HttpHeader): ...\n",
    "\n",
    "async def f(req):\n",
    "    a = await _wrap_req(req, _params(g))\n",
    "    return Response(str(a))\n",
    "\n",
    "client = TestClient(Starlette(routes=[Route('/', f, methods=['POST'])]))\n",
    "response = client.post('/?a=1', data=d)\n",
    "print(response.text)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "34066c89",
   "metadata": {},
   "source": [
    "**Missing Request Params**"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "06259883",
   "metadata": {},
   "source": [
    "If a request param has a default value (e.g. `a:str=''`), the request is valid even if the user doesn't include the param in their request."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "id": "c369a89c",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{'req': <starlette.requests.Request object at 0x11124b860>, 'this': <starlette.applications.Starlette object at 0x110250470>, 'a': ''}\n"
     ]
    }
   ],
   "source": [
    "def g(req, this:Starlette, a:str=''): ...\n",
    "\n",
    "async def f(req):\n",
    "    a = await _wrap_req(req, _params(g))\n",
    "    return Response(str(a))\n",
    "\n",
    "client = TestClient(Starlette(routes=[Route('/', f, methods=['POST'])]))\n",
    "response = client.post('/', json={}) # no param in request\n",
    "print(response.text)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "35b68f4f",
   "metadata": {},
   "source": [
    "If we remove the default value and re-run the request, we should get the following error `Missing required field: a`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "id": "b33b43a7",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Missing required field: a\n"
     ]
    }
   ],
   "source": [
    "def g(req, this:Starlette, a:str): ...\n",
    "\n",
    "async def f(req):\n",
    "    a = await _wrap_req(req, _params(g))\n",
    "    return Response(str(a))\n",
    "\n",
    "client = TestClient(Starlette(routes=[Route('/', f, methods=['POST'])]))\n",
    "response = client.post('/', json={}) # no param in request\n",
    "print(response.text)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "id": "7a661bfa",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def flat_xt(lst):\n",
    "    \"Flatten lists\"\n",
    "    result = []\n",
    "    if isinstance(lst,(FT,str)): lst=[lst]\n",
    "    for item in lst:\n",
    "        if isinstance(item, (list,tuple)): result.extend(item)\n",
    "        else: result.append(item)\n",
    "    return tuple(result)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "id": "2ee5adf1",
   "metadata": {},
   "outputs": [],
   "source": [
    "x = ft('a',1)\n",
    "test_eq(flat_xt([x, x, [x,x]]), (x,)*4)\n",
    "test_eq(flat_xt(x), (x,))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "id": "aacff5ac",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "class Beforeware:\n",
    "    def __init__(self, f, skip=None): self.f,self.skip = f,skip or []"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "id": "78c3c357",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "async def _handle(f, *args, **kwargs):\n",
    "    return (await f(*args, **kwargs)) if is_async_callable(f) else await run_in_threadpool(f, *args, **kwargs)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "079a3215",
   "metadata": {},
   "source": [
    "## Websockets / SSE"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "id": "f2277c02",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def _find_wsp(ws, data, hdrs, arg:str, p:Parameter):\n",
    "    \"In `data` find param named `arg` of type in `p` (`arg` is ignored for body types)\"\n",
    "    anno = p.annotation\n",
    "    if isinstance(anno, type):\n",
    "        if issubclass(anno, HtmxHeaders): return _get_htmx(hdrs)\n",
    "        if issubclass(anno, Starlette): return ws.scope['app']\n",
    "        if issubclass(anno, WebSocket): return ws\n",
    "        if issubclass(anno, dict): return data\n",
    "    if anno is empty:\n",
    "        if arg.lower()=='ws': return ws\n",
    "        if arg.lower()=='scope': return dict2obj(ws.scope)\n",
    "        if arg.lower()=='data': return data\n",
    "        if arg.lower()=='htmx': return _get_htmx(hdrs)\n",
    "        if arg.lower()=='app': return ws.scope['app']\n",
    "        if arg.lower()=='send': return partial(_send_ws, ws)\n",
    "        if 'session'.startswith(arg.lower()): return ws.scope.get('session', {})\n",
    "        return None\n",
    "    res = data.get(arg, None)\n",
    "    if res is empty or res is None: res = hdrs.get(arg, None)\n",
    "    if res is empty or res is None: res = p.default\n",
    "    # We can cast str and list[str] to types; otherwise just return what we have\n",
    "    if not isinstance(res, (list,str)) or anno is empty: return res\n",
    "    return [_fix_anno(anno, o) for o in res] if isinstance(res,list) else _fix_anno(anno, res)\n",
    "\n",
    "def _wrap_ws(ws, data, params):\n",
    "    hdrs = {k.lower().replace('-','_'):v for k,v in data.pop('HEADERS', {}).items()}\n",
    "    return {arg:_find_wsp(ws, data, hdrs, arg, p) for arg,p in params.items()}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "id": "dcc15129",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "async def _send_ws(ws, resp):\n",
    "    if not resp: return\n",
    "    res = to_xml(resp, indent=fh_cfg.indent)\n",
    "    await ws.send_text(res)\n",
    "\n",
    "def _ws_endp(recv, conn=None, disconn=None):\n",
    "    cls = type('WS_Endp', (WebSocketEndpoint,), {\"encoding\":\"text\"})\n",
    "\n",
    "    async def _generic_handler(handler, ws, data=None):\n",
    "        wd = _wrap_ws(ws, loads(data) if data else {}, _params(handler))\n",
    "        resp = await _handle(handler, **wd)\n",
    "        if resp: await _send_ws(ws, resp)\n",
    "\n",
    "    async def _connect(self, ws):\n",
    "        await ws.accept()\n",
    "        await _generic_handler(conn, ws)\n",
    "\n",
    "    async def _disconnect(self, ws, close_code): await _generic_handler(disconn, ws)\n",
    "    async def _recv(self, ws, data): await _generic_handler(recv, ws, data)\n",
    "\n",
    "    if    conn: cls.on_connect    = _connect\n",
    "    if disconn: cls.on_disconnect = _disconnect\n",
    "    cls.on_receive = _recv\n",
    "    return cls"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "id": "983bcfe2",
   "metadata": {},
   "outputs": [],
   "source": [
    "def on_receive(self, msg:str): return f\"Message text was: {msg}\"\n",
    "c = _ws_endp(on_receive)\n",
    "cli = TestClient(Starlette(routes=[WebSocketRoute('/', _ws_endp(on_receive))]))\n",
    "with cli.websocket_connect('/') as ws:\n",
    "    ws.send_text('{\"msg\":\"Hi!\"}')\n",
    "    data = ws.receive_text()\n",
    "    assert data == 'Message text was: Hi!'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "id": "5b0e7677",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def EventStream(s):\n",
    "    \"Create a text/event-stream response from `s`\"\n",
    "    return StreamingResponse(s, media_type=\"text/event-stream\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 52,
   "id": "0dd0a414",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def signal_shutdown():\n",
    "    from uvicorn.main import Server\n",
    "    event = asyncio.Event()\n",
    "    @patch\n",
    "    def handle_exit(self:Server, *args, **kwargs):\n",
    "        event.set()\n",
    "        self.force_exit = True\n",
    "        self._orig_handle_exit(*args, **kwargs)\n",
    "    return event"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "496cd78f",
   "metadata": {},
   "source": [
    "## Routing and application"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "id": "03f4c639",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def uri(_arg, **kwargs):\n",
    "    \"Create a URI by URL-encoding `_arg` and appending query parameters from `kwargs`\"\n",
    "    return f\"{quote(_arg)}/{urlencode(kwargs, doseq=True)}\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 54,
   "id": "bdb2ac14",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def decode_uri(s):\n",
    "    \"Decode a URI created by `uri()` back into argument and keyword dict\"\n",
    "    arg,_,kw = s.partition('/')\n",
    "    return unquote(arg), {k:v[0] for k,v in parse_qs(kw).items()}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "id": "b80ce139",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "from starlette.convertors import StringConvertor"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 56,
   "id": "a27adc1e",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "StringConvertor.regex = \"[^/]*\"  # `+` replaced with `*`\n",
    "\n",
    "@patch\n",
    "def to_string(self:StringConvertor, value: str) -> str:\n",
    "    value = str(value)\n",
    "    assert \"/\" not in value, \"May not contain path separators\"\n",
    "    # assert value, \"Must not be empty\"  # line removed due to errors\n",
    "    return value"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 57,
   "id": "46614165",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "@patch\n",
    "def url_path_for(self:HTTPConnection, name: str, **path_params):\n",
    "    lp = self.scope['app'].url_path_for(name, **path_params)\n",
    "    return URLPath(f\"{self.scope['root_path']}{lp}\", lp.protocol, lp.host)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 58,
   "id": "c1707d59",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "_verbs = dict(get='hx-get', post='hx-post', put='hx-post', delete='hx-delete', patch='hx-patch', link='href')\n",
    "\n",
    "def _url_for(req, t):\n",
    "    \"Generate URL for route `t` using request `req`\"\n",
    "    if callable(t): t = t.__routename__\n",
    "    kw = {}\n",
    "    if t.find('/')>-1 and (t.find('?')<0 or t.find('/')<t.find('?')): t,kw = decode_uri(t)\n",
    "    t,m,q = t.partition('?')\n",
    "    return f\"{req.url_path_for(t, **kw)}{m}{q}\"\n",
    "\n",
    "def _find_targets(req, resp):\n",
    "    \"Find and convert route targets in response attributes to URLs\"\n",
    "    if isinstance(resp, tuple):\n",
    "        for o in resp: _find_targets(req, o)\n",
    "    if isinstance(resp, FT):\n",
    "        for o in resp.children: _find_targets(req, o)\n",
    "        for k,v in _verbs.items():\n",
    "            t = resp.attrs.pop(k, None)\n",
    "            if t: resp.attrs[v] = _url_for(req, t)\n",
    "\n",
    "def _apply_ft(o):\n",
    "    \"Apply FastTag transformation recursively to object `o`\"\n",
    "    if isinstance(o, tuple): o = tuple(_apply_ft(c) for c in o)\n",
    "    if hasattr(o, '__ft__'): o = o.__ft__()\n",
    "    if isinstance(o, FT): o.children = tuple(_apply_ft(c) for c in o.children)\n",
    "    return o\n",
    "\n",
    "def _to_xml(req, resp, indent):\n",
    "    \"Convert response to XML string with target URL resolution\"\n",
    "    resp = _apply_ft(resp)\n",
    "    _find_targets(req, resp)\n",
    "    return to_xml(resp, indent)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 59,
   "id": "f1e3ed2d",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "_iter_typs = (tuple,list,map,filter,range,types.GeneratorType)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 60,
   "id": "a407dc0e",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def flat_tuple(o):\n",
    "    \"Flatten nested iterables into a single tuple\"\n",
    "    result = []\n",
    "    if not isinstance(o,_iter_typs): o=[o]\n",
    "    o = list(o)\n",
    "    for item in o:\n",
    "        if isinstance(item, _iter_typs): result.extend(list(item))\n",
    "        else: result.append(item)\n",
    "    return tuple(result)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 61,
   "id": "d36402e6",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def noop_body(c, req):\n",
    "    \"Default Body wrap function which just returns the content\"\n",
    "    return c"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 62,
   "id": "7f49728d",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def respond(req, heads, bdy):\n",
    "    \"Default FT response creation function\"\n",
    "    body_wrap = getattr(req, 'body_wrap', noop_body)\n",
    "    params = inspect.signature(body_wrap).parameters\n",
    "    bw_args = (bdy, req) if len(params)>1 else (bdy,)\n",
    "    body = Body(body_wrap(*bw_args), *flat_xt(req.ftrs), **req.bodykw)\n",
    "    return Html(Head(*heads, *flat_xt(req.hdrs)), body, **req.htmlkw)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "cfae0a56",
   "metadata": {},
   "source": [
    "Render fragment if `HX-Request` header is _present_ and\n",
    "  `HX-History-Restore-Request` header is _absent._"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 63,
   "id": "b2007479",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def is_full_page(req, resp):\n",
    "    \"Check if response should be rendered as full page or fragment\"\n",
    "    if resp and any(getattr(o, 'tag', '')=='html' for o in resp): return True\n",
    "    return 'hx-request' in req.headers and 'hx-history-restore-request' not in req.headers"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 64,
   "id": "358badec",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def _part_resp(req, resp):\n",
    "    \"Partition response into HTTP headers, background tasks, and content\"\n",
    "    resp = flat_tuple(resp)\n",
    "    resp = resp + tuple(getattr(req, 'injects', ()))\n",
    "    http_hdrs,resp = partition(resp, risinstance(HttpHeader))\n",
    "    tasks,resp = partition(resp, risinstance(BackgroundTask))\n",
    "    kw = {\"headers\": {\"vary\": \"HX-Request, HX-History-Restore-Request\"}}\n",
    "    if http_hdrs: kw['headers'] |= {o.k:str(o.v) for o in http_hdrs}\n",
    "    if tasks:\n",
    "        ts = BackgroundTasks()\n",
    "        for t in tasks: ts.tasks.append(t)\n",
    "        kw['background'] = ts\n",
    "    resp = tuple(resp)\n",
    "    if len(resp)==1: resp = resp[0]\n",
    "    return resp,kw"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 65,
   "id": "da449a7b",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def _canonical(req):\n",
    "    if not req.app.canonical: return []\n",
    "    url = str(getattr(req, 'canonical', req.url)).replace('http://', 'https://', 1)\n",
    "    return [Link(rel=\"canonical\", href=url)]\n",
    "\n",
    "def _xt_cts(req, resp):\n",
    "    \"Extract content and headers, render as full page or fragment\"\n",
    "    hdr_tags = 'title','meta','link','style','base'\n",
    "    resp = tuplify(resp)\n",
    "    heads,bdy = partition(resp, lambda o: getattr(o, 'tag', '') in hdr_tags)\n",
    "    if not is_full_page(req, resp):\n",
    "        title = [] if any(getattr(o, 'tag', '')=='title' for o in heads) else [Title(req.app.title)]\n",
    "        resp = respond(req, [*heads, *title, *_canonical(req)], bdy)\n",
    "    return _to_xml(req, resp, indent=fh_cfg.indent)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 66,
   "id": "775fb66b",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def _is_ft_resp(resp):\n",
    "    \"Check if response is a FastTag-compatible type\"\n",
    "    return isinstance(resp, _iter_typs+(HttpHeader,FT)) or hasattr(resp, '__ft__')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 67,
   "id": "968d9245",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def _resp(req, resp, cls=empty, status_code=200):\n",
    "    \"Create appropriate HTTP response from request and response data\"\n",
    "    if not resp: resp=''\n",
    "    if hasattr(resp, '__response__'): resp = resp.__response__(req)\n",
    "    if cls in (Any,FT): cls=empty\n",
    "    if isinstance(resp, FileResponse) and not os.path.exists(resp.path): raise HTTPException(404, resp.path)\n",
    "    resp,kw = _part_resp(req, resp)\n",
    "    if isinstance(resp, Response): return resp\n",
    "    if cls is not empty: return cls(resp, status_code=status_code, **kw)\n",
    "    if _is_ft_resp(resp):\n",
    "        cts = _xt_cts(req, resp)\n",
    "        return HTMLResponse(cts, status_code=status_code, **kw)\n",
    "    if isinstance(resp, str): cls = HTMLResponse\n",
    "    elif isinstance(resp, Mapping): cls = JSONResponse\n",
    "    else:\n",
    "        resp = str(resp)\n",
    "        cls = HTMLResponse\n",
    "    return cls(resp, status_code=status_code, **kw)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 68,
   "id": "eac44461",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "class Redirect:\n",
    "    \"Use HTMX or Starlette RedirectResponse as required to redirect to `loc`\"\n",
    "    def __init__(self, loc): self.loc = loc\n",
    "    def __response__(self, req):\n",
    "        if 'hx-request' in req.headers: return HtmxResponseHeaders(redirect=self.loc)\n",
    "        return RedirectResponse(self.loc, status_code=303)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 69,
   "id": "1ba52822",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "async def _wrap_call(f, req, params):\n",
    "    \"Wrap function call with request parameter injection\"\n",
    "    wreq = await _wrap_req(req, params)\n",
    "    return await _handle(f, **wreq)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 70,
   "id": "b0d1cbbf",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "htmx_exts = {\n",
    "    \"morph\": \"https://cdn.jsdelivr.net/npm/idiomorph@0.7.3/dist/idiomorph-ext.min.js\",\n",
    "    \"head-support\": \"https://cdn.jsdelivr.net/npm/htmx-ext-head-support@2.0.4/head-support.js\",\n",
    "    \"preload\": \"https://cdn.jsdelivr.net/npm/htmx-ext-preload@2.1.1/preload.js\",\n",
    "    \"class-tools\": \"https://cdn.jsdelivr.net/npm/htmx-ext-class-tools@2.0.1/class-tools.js\",\n",
    "    \"loading-states\": \"https://cdn.jsdelivr.net/npm/htmx-ext-loading-states@2.0.1/loading-states.js\",\n",
    "    \"multi-swap\": \"https://cdn.jsdelivr.net/npm/htmx-ext-multi-swap@2.0.0/multi-swap.js\",\n",
    "    \"path-deps\": \"https://cdn.jsdelivr.net/npm/htmx-ext-path-deps@2.0.0/path-deps.js\",\n",
    "    \"remove-me\": \"https://cdn.jsdelivr.net/npm/htmx-ext-remove-me@2.0.0/remove-me.js\",\n",
    "    \"debug\": \"https://unpkg.com/htmx.org@1.9.12/dist/ext/debug.js\",\n",
    "    \"ws\": \"https://cdn.jsdelivr.net/npm/htmx-ext-ws@2.0.3/ws.js\",\n",
    "    \"chunked-transfer\": \"https://cdn.jsdelivr.net/npm/htmx-ext-transfer-encoding-chunked@0.4.0/transfer-encoding-chunked.js\"\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3531273d",
   "metadata": {},
   "source": [
    "The FastHTML `exts` param supports the following:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 71,
   "id": "deffbaaa",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "morph head-support preload class-tools loading-states multi-swap path-deps remove-me debug ws chunked-transfer\n"
     ]
    }
   ],
   "source": [
    "print(' '.join(htmx_exts))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 72,
   "id": "60cb52ea",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "htmxsrc   = Script(src=\"https://cdn.jsdelivr.net/npm/htmx.org@2.0.7/dist/htmx.js\")\n",
    "fhjsscr   = Script(src=\"https://cdn.jsdelivr.net/gh/answerdotai/fasthtml-js@1.0.12/fasthtml.js\")\n",
    "surrsrc   = Script(src=\"https://cdn.jsdelivr.net/gh/answerdotai/surreal@main/surreal.js\")\n",
    "scopesrc  = Script(src=\"https://cdn.jsdelivr.net/gh/gnat/css-scope-inline@main/script.js\")\n",
    "viewport  = Meta(name=\"viewport\", content=\"width=device-width, initial-scale=1, viewport-fit=cover\")\n",
    "charset   = Meta(charset=\"utf-8\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 73,
   "id": "8bd78eeb",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def get_key(key=None, fname='.sesskey'):\n",
    "    \"Get session key from `key` param or read/create from file `fname`\"\n",
    "    if key: return key\n",
    "    fname = Path(fname)\n",
    "    if fname.exists(): return fname.read_text()\n",
    "    key = str(uuid.uuid4())\n",
    "    fname.write_text(key)\n",
    "    return key"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 74,
   "id": "042666e2",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def _list(o):\n",
    "    \"Wrap non-list item in a list, returning empty list if None\"\n",
    "    return [] if not o else list(o) if isinstance(o, (tuple,list)) else [o]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 75,
   "id": "d276fc71",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def _wrap_ex(f, status_code, hdrs, ftrs, htmlkw, bodykw, body_wrap):\n",
    "    \"Wrap exception handler with FastHTML request processing\"\n",
    "    async def _f(req, exc):\n",
    "        req.hdrs,req.ftrs,req.htmlkw,req.bodykw = map(deepcopy, (hdrs, ftrs, htmlkw, bodykw))\n",
    "        req.body_wrap = body_wrap\n",
    "        res = await _handle(f, req, exc)\n",
    "        return _resp(req, res, status_code=status_code)\n",
    "    return _f"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 76,
   "id": "bc323fd4",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def qp(p:str, **kw) -> str:\n",
    "    \"Add parameters kw to path p\"\n",
    "    def _sub(m):\n",
    "        pre,post = m.groups()\n",
    "        if pre not in kw: return f'{{{pre}{post or \"\"}}}'\n",
    "        pre = kw.pop(pre)\n",
    "        return '' if pre in (False,None) else str(pre)\n",
    "    p = re.sub(r'\\{([^:}]+)(:.+?)?}', _sub, p)\n",
    "    # encode query params\n",
    "    return p + ('?' + urlencode({k:'' if v in (False,None) else v for k,v in kw.items()},doseq=True) if kw else '')"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e51e2dc3",
   "metadata": {},
   "source": [
    "`qp` adds query parameters to route path strings"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 77,
   "id": "ff82dc78",
   "metadata": {},
   "outputs": [],
   "source": [
    "vals = {'a':5, 'b':False, 'c':[1,2], 'd':'bar', 'e':None, 'ab':42}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 78,
   "id": "c0836a8f",
   "metadata": {},
   "outputs": [],
   "source": [
    "res = qp('/foo', **vals)\n",
    "test_eq(res, '/foo?a=5&b=&c=1&c=2&d=bar&e=&ab=42')"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4740272f",
   "metadata": {},
   "source": [
    "`qp` checks to see if each param should be sent as a query parameter or as part of the route, and encodes that properly."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 79,
   "id": "50ebb1ee",
   "metadata": {},
   "outputs": [],
   "source": [
    "path = '/foo/{a}/{d}/{ab:int}'\n",
    "res = qp(path, **vals)\n",
    "test_eq(res, '/foo/5/bar/42?b=&c=1&c=2&e=')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 80,
   "id": "f86690c4",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def def_hdrs(htmx=True, surreal=True):\n",
    "    \"Default headers for a FastHTML app\"\n",
    "    hdrs = []\n",
    "    if surreal: hdrs = [surrsrc,scopesrc] + hdrs\n",
    "    if htmx: hdrs = [htmxsrc,fhjsscr] + hdrs\n",
    "    return [charset, viewport] + hdrs"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 81,
   "id": "2c5285ae",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "cors_allow = Middleware(CORSMiddleware, allow_credentials=True,\n",
    "                        allow_origins=[\"*\"], allow_methods=[\"*\"], allow_headers=[\"*\"])\n",
    "\n",
    "iframe_scr = Script(NotStr(\"\"\"\n",
    "    function sendmsg() {\n",
    "        window.parent.postMessage({height: document.documentElement.offsetHeight}, '*');\n",
    "    }\n",
    "    window.onload = function() {\n",
    "        sendmsg();\n",
    "        document.body.addEventListener('htmx:afterSettle',    sendmsg);\n",
    "        document.body.addEventListener('htmx:wsAfterMessage', sendmsg);\n",
    "    };\"\"\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 82,
   "id": "3327a1e9",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "class FastHTML(Starlette):\n",
    "    def __init__(self, debug=False, routes=None, middleware=None, title: str = \"FastHTML page\", exception_handlers=None,\n",
    "                 on_startup=None, on_shutdown=None, lifespan=None, hdrs=None, ftrs=None, exts=None,\n",
    "                 before=None, after=None, surreal=True, htmx=True, default_hdrs=True, sess_cls=SessionMiddleware,\n",
    "                 secret_key=None, session_cookie='session_', max_age=365*24*3600, sess_path='/',\n",
    "                 same_site='lax', sess_https_only=False, sess_domain=None, key_fname='.sesskey',\n",
    "                 body_wrap=noop_body, htmlkw=None, nb_hdrs=False, canonical=True, **bodykw):\n",
    "        middleware,before,after = map(_list, (middleware,before,after))\n",
    "        self.title,self.canonical,self.session_cookie,self.key_fname = title,canonical,session_cookie,key_fname\n",
    "        hdrs,ftrs,exts = map(listify, (hdrs,ftrs,exts))\n",
    "        exts = {k:htmx_exts[k] for k in exts}\n",
    "        htmlkw = htmlkw or {}\n",
    "        if default_hdrs: hdrs = def_hdrs(htmx, surreal=surreal) + hdrs\n",
    "        hdrs += [Script(src=ext) for ext in exts.values()]\n",
    "        if IN_NOTEBOOK:\n",
    "            hdrs.append(iframe_scr)\n",
    "            from IPython.display import display,HTML\n",
    "            if nb_hdrs: display(HTML(to_xml(tuple(hdrs))))\n",
    "            middleware.append(cors_allow)\n",
    "        on_startup,on_shutdown = listify(on_startup) or None,listify(on_shutdown) or None\n",
    "        self.lifespan,self.hdrs,self.ftrs = lifespan,hdrs,ftrs\n",
    "        self.body_wrap,self.before,self.after,self.htmlkw,self.bodykw = body_wrap,before,after,htmlkw,bodykw\n",
    "        self.secret_key = get_key(secret_key, key_fname)\n",
    "        if sess_cls:\n",
    "            sess = Middleware(sess_cls, secret_key=self.secret_key,session_cookie=session_cookie,\n",
    "                              max_age=max_age, path=sess_path, same_site=same_site,\n",
    "                              https_only=sess_https_only, domain=sess_domain)\n",
    "            middleware.append(sess)\n",
    "        exception_handlers = ifnone(exception_handlers, {})\n",
    "        if 404 not in exception_handlers:\n",
    "            def _not_found(req, exc): return  Response('404 Not Found', status_code=404)\n",
    "            exception_handlers[404] = _not_found\n",
    "        excs = {k:_wrap_ex(v, k, hdrs, ftrs, htmlkw, bodykw, body_wrap=body_wrap) for k,v in exception_handlers.items()}\n",
    "        super().__init__(debug, routes, middleware=middleware, exception_handlers=excs, on_startup=on_startup, on_shutdown=on_shutdown, lifespan=lifespan)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 83,
   "id": "e0accf76",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "@patch\n",
    "def add_route(self:FastHTML, route):\n",
    "    \"Add or replace a route in the FastHTML app\"\n",
    "    route.methods = [m.upper() for m in listify(route.methods)]\n",
    "    self.router.routes = [r for r in self.router.routes if not\n",
    "                   (r.path==route.path and r.name == route.name and\n",
    "                    ((route.methods is None) or (set(r.methods) == set(route.methods))))]\n",
    "    self.router.routes.append(route)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 84,
   "id": "246bd8d1",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "all_meths = 'get post put delete patch head trace options'.split()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 85,
   "id": "26b147ba",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "@patch\n",
    "def _endp(self:FastHTML, f, body_wrap):\n",
    "    \"Create endpoint wrapper with before/after middleware processing\"\n",
    "    sig = signature_ex(f, True)\n",
    "    async def _f(req):\n",
    "        resp = None\n",
    "        req.injects = []\n",
    "        req.hdrs,req.ftrs,req.htmlkw,req.bodykw = map(deepcopy, (self.hdrs,self.ftrs,self.htmlkw,self.bodykw))\n",
    "        req.hdrs,req.ftrs = listify(req.hdrs),listify(req.ftrs)\n",
    "        for b in self.before:\n",
    "            if not resp:\n",
    "                if isinstance(b, Beforeware): bf,skip = b.f,b.skip\n",
    "                else: bf,skip = b,[]\n",
    "                if not any(re.fullmatch(r, req.url.path) for r in skip):\n",
    "                    resp = await _wrap_call(bf, req, _params(bf))\n",
    "        req.body_wrap = body_wrap\n",
    "        if not resp: resp = await _wrap_call(f, req, sig.parameters)\n",
    "        for a in self.after:\n",
    "            wreq = await _wrap_req(req, _params(a))\n",
    "            wreq['resp'] = resp\n",
    "            nr = a(**wreq)\n",
    "            if nr: resp = nr\n",
    "        return _resp(req, resp, sig.return_annotation)\n",
    "    return _f"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 86,
   "id": "3818575c",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "@patch\n",
    "def _add_ws(self:FastHTML, func, path, conn, disconn, name, middleware):\n",
    "    \"Add websocket route to FastHTML app\"\n",
    "    endp = _ws_endp(func, conn, disconn)\n",
    "    route = WebSocketRoute(path, endpoint=endp, name=name, middleware=middleware)\n",
    "    route.methods = ['ws']\n",
    "    self.add_route(route)\n",
    "    return func"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 87,
   "id": "669e76eb",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "@patch\n",
    "def ws(self:FastHTML, path:str, conn=None, disconn=None, name=None, middleware=None):\n",
    "    \"Add a websocket route at `path`\"\n",
    "    def f(func=noop): return self._add_ws(func, path, conn, disconn, name=name, middleware=middleware)\n",
    "    return f"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 88,
   "id": "919618c3",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def _mk_locfunc(f, p, app=None):\n",
    "    \"Create a location function wrapper with route path and to() method\"\n",
    "    class _lf:\n",
    "        def __init__(self):\n",
    "            update_wrapper(self, f)\n",
    "            self.app = app\n",
    "\n",
    "        def __call__(self, *args, **kw): return f(*args, **kw)\n",
    "        def to(self, **kw): return qp(p, **kw)\n",
    "        def __str__(self): return p\n",
    "\n",
    "    return _lf()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 89,
   "id": "d2ecc738",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def nested_name(f):\n",
    "    \"Get name of function `f` using '_' to join nested function names\"\n",
    "    return f.__qualname__.replace('.<locals>.', '_')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 90,
   "id": "c0f13ece",
   "metadata": {},
   "outputs": [],
   "source": [
    "def f():\n",
    "    def g(): ...\n",
    "    return g"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 91,
   "id": "b218f738",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'f_g'"
      ]
     },
     "execution_count": 91,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "func = f()\n",
    "nested_name(func)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 92,
   "id": "72760b09",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "@patch\n",
    "def _add_route(self:FastHTML, func, path, methods, name, include_in_schema, body_wrap):\n",
    "    \"Add HTTP route to FastHTML app with automatic method detection\"\n",
    "    n,fn,p = name,nested_name(func),None if callable(path) else path\n",
    "    if methods: m = [methods] if isinstance(methods,str) else methods\n",
    "    elif fn in all_meths and p is not None: m = [fn]\n",
    "    else: m = ['get','post']\n",
    "    if not n: n = fn\n",
    "    if not p: p = '/'+('' if fn=='index' else fn)\n",
    "    route = Route(p, endpoint=self._endp(func, body_wrap or self.body_wrap), methods=m, name=n, include_in_schema=include_in_schema)\n",
    "    self.add_route(route)\n",
    "    lf = _mk_locfunc(func, p, app=self)\n",
    "    lf.__routename__ = n\n",
    "    return lf"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 93,
   "id": "f5cb2c2b",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "@patch\n",
    "def route(self:FastHTML, path:str=None, methods=None, name=None, include_in_schema=True, body_wrap=None):\n",
    "    \"Add a route at `path`\"\n",
    "    def f(func): return self._add_route(func, path, methods, name=name, include_in_schema=include_in_schema, body_wrap=body_wrap)\n",
    "    return f(path) if callable(path) else f\n",
    "\n",
    "for o in all_meths: setattr(FastHTML, o, partialmethod(FastHTML.route, methods=o))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 94,
   "id": "e6ee3a86",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'/foo?a=bar&b=1&b=2'"
      ]
     },
     "execution_count": 94,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "app = FastHTML()\n",
    "@app.get\n",
    "def foo(a:str, b:list[int]): ...\n",
    "\n",
    "foo.to(a='bar', b=[1,2])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 95,
   "id": "9b9f1f03",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'/foo/bar?b=1&b=2'"
      ]
     },
     "execution_count": 95,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "@app.get('/foo/{a}')\n",
    "def foo(a:str, b:list[int]): ...\n",
    "\n",
    "foo.to(a='bar', b=[1,2])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 96,
   "id": "35c35a96",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "@patch\n",
    "def set_lifespan(self:FastHTML, value):\n",
    "    \"Set the lifespan context manager for the FastHTML app\"\n",
    "    if inspect.isasyncgenfunction(value): value = contextlib.asynccontextmanager(value)\n",
    "    self.router.lifespan_context = value"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 97,
   "id": "3a348474",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def serve(\n",
    "        appname=None, # Name of the module\n",
    "        app='app', # App instance to be served\n",
    "        host='0.0.0.0', # If host is 0.0.0.0 will convert to localhost\n",
    "        port=None, # If port is None it will default to 5001 or the PORT environment variable\n",
    "        reload=True, # Default is to reload the app upon code changes\n",
    "        reload_includes:list[str]|str|None=None, # Additional files to watch for changes\n",
    "        reload_excludes:list[str]|str|None=None # Files to ignore for changes\n",
    "        ):\n",
    "    \"Run the app in an async server, with live reload set as the default.\"\n",
    "    bk = inspect.currentframe().f_back\n",
    "    glb = bk.f_globals\n",
    "    code = bk.f_code\n",
    "    if not appname:\n",
    "        if glb.get('__name__')=='__main__': appname = Path(glb.get('__file__', '')).stem\n",
    "        elif code.co_name=='main' and bk.f_back.f_globals.get('__name__')=='__main__': appname = inspect.getmodule(bk).__name__\n",
    "    import uvicorn\n",
    "    if appname:\n",
    "        if not port: port=int(os.getenv(\"PORT\", default=5001))\n",
    "        print(f'Link: http://{\"localhost\" if host==\"0.0.0.0\" else host}:{port}')\n",
    "        uvicorn.run(f'{appname}:{app}', host=host, port=port, reload=reload, reload_includes=reload_includes, reload_excludes=reload_excludes)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 98,
   "id": "8121968a",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "class Client:\n",
    "    \"A simple httpx ASGI client that doesn't require `async`\"\n",
    "    def __init__(self, app, url=\"http://testserver\"):\n",
    "        self.cli = httpx.AsyncClient(transport=httpx.ASGITransport(app), base_url=url)\n",
    "\n",
    "    def _sync(self, method, url, **kwargs):\n",
    "        async def _request(): return await self.cli.request(method, url, **kwargs)\n",
    "        with from_thread.start_blocking_portal() as portal: return portal.call(_request)\n",
    "\n",
    "for o in ('get', 'post', 'delete', 'put', 'patch', 'options'): setattr(Client, o, partialmethod(Client._sync, o))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 99,
   "id": "b163c933",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'test'"
      ]
     },
     "execution_count": 99,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "app = FastHTML(routes=[Route('/', lambda _: Response('test'))])\n",
    "cli = Client(app)\n",
    "\n",
    "cli.get('/').text"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "efc8e6db",
   "metadata": {},
   "source": [
    "Note that you can also use Starlette's `TestClient` instead of FastHTML's `Client`. They should be largely interchangable."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2d2633dd",
   "metadata": {},
   "source": [
    "## FastHTML Tests"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 100,
   "id": "9abc3781",
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_cli(app): return app,TestClient(app),app.route"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 101,
   "id": "645d8d95",
   "metadata": {},
   "outputs": [],
   "source": [
    "app,cli,rt = get_cli(FastHTML(secret_key='soopersecret'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 102,
   "id": "421262a8",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[Route(path='/foo', name='foo', methods=['GET', 'HEAD'])]\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "'/foo?param=value'"
      ]
     },
     "execution_count": 102,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "app,cli,rt = get_cli(FastHTML(title=\"My Custom Title\"))\n",
    "@app.get\n",
    "def foo(): return Div(\"Hello World\")\n",
    "\n",
    "print(app.routes)\n",
    "\n",
    "response = cli.get('/foo')\n",
    "assert '<title>My Custom Title</title>' in response.text\n",
    "\n",
    "foo.to(param='value')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2ebd6270",
   "metadata": {},
   "outputs": [],
   "source": [
    "app,cli,rt = get_cli(FastHTML())\n",
    "\n",
    "@rt('/xt2')\n",
    "def get(): return H1('bar')\n",
    "\n",
    "txt = cli.get('/xt2').text\n",
    "assert '<title>FastHTML page</title>' in txt and '<h1>bar</h1>' in txt and '<html>' in txt"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8a6ed879",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'Hi there'"
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "@rt(\"/hi\")\n",
    "def get(): return 'Hi there'\n",
    "\n",
    "r = cli.get('/hi')\n",
    "r.text"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a8e723fc",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'Postal'"
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "@rt(\"/hi\")\n",
    "def post(): return 'Postal'\n",
    "\n",
    "cli.post('/hi').text"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "274666db",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'testserver'"
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "@app.get(\"/hostie\")\n",
    "def show_host(req): return req.headers['host']\n",
    "\n",
    "cli.get('/hostie').text"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "72428702",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Message text was: Hi! with session bar, from client: Address(host='testclient', port=50000)\n"
     ]
    }
   ],
   "source": [
    "@app.get(\"/setsess\")\n",
    "def set_sess(session):\n",
    "   session['foo'] = 'bar'\n",
    "   return 'ok'\n",
    "\n",
    "@app.ws(\"/ws\")\n",
    "def ws(self, msg:str, ws:WebSocket, session): return f\"Message text was: {msg} with session {session.get('foo')}, from client: {ws.client}\"\n",
    "\n",
    "cli.get('/setsess')\n",
    "with cli.websocket_connect('/ws') as ws:\n",
    "    ws.send_text('{\"msg\":\"Hi!\"}')\n",
    "    data = ws.receive_text()\n",
    "assert 'Message text was: Hi! with session bar' in data\n",
    "print(data)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "36292bf3",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'a yoyo'"
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "@rt\n",
    "def yoyo(): return 'a yoyo'\n",
    "\n",
    "cli.post('/yoyo').text"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0d813cd0",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      " <!doctype html>\n",
      " <html>\n",
      "   <div hx-post=\"a yoyo\">Text.</div>\n",
      " </html>\n",
      "\n"
     ]
    }
   ],
   "source": [
    "@app.get\n",
    "def autopost(): return Html(Div('Text.', hx_post=yoyo()))\n",
    "print(cli.get('/autopost').text)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "52dde0da",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      " <!doctype html>\n",
      " <html>\n",
      "   <body>\n",
      "     <div class=\"px-2\" hx-post=\"/hostie?a=b\">Text.</div>\n",
      "   </body>\n",
      " </html>\n",
      "\n"
     ]
    }
   ],
   "source": [
    "@app.get\n",
    "def autopost2(): return Html(Body(Div('Text.', cls='px-2', hx_post=show_host.to(a='b'))))\n",
    "print(cli.get('/autopost2').text)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d84d98f1",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      " <!doctype html>\n",
      " <html>\n",
      "   <div hx-get=\"/hostie\">Text.</div>\n",
      " </html>\n",
      "\n"
     ]
    }
   ],
   "source": [
    "@app.get\n",
    "def autoget2(): return Html(Div('Text.', hx_get=show_host))\n",
    "print(cli.get('/autoget2').text)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "64343367",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'Good day to you, Alexis!'"
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "@rt('/user/{nm}', name='gday')\n",
    "def get(nm:str=''): return f\"Good day to you, {nm}!\"\n",
    "cli.get('/user/Alexis').text"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "29a96715",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      " <!doctype html>\n",
      " <html>\n",
      "   <div href=\"/user/Alexis\">Text.</div>\n",
      " </html>\n",
      "\n"
     ]
    }
   ],
   "source": [
    "@app.get\n",
    "def autolink(): return Html(Div('Text.', link=uri('gday', nm='Alexis')))\n",
    "print(cli.get('/autolink').text)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "54266599",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'http://testserver/user/Alexis; http://testserver/hostie'"
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "@rt('/link')\n",
    "def get(req): return f\"{req.url_for('gday', nm='Alexis')}; {req.url_for('show_host')}\"\n",
    "\n",
    "cli.get('/link').text"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "162b811e",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Background task completed!\n"
     ]
    }
   ],
   "source": [
    "@app.get(\"/background\")\n",
    "async def background_task(request):\n",
    "    async def long_running_task():\n",
    "        await asyncio.sleep(0.1)\n",
    "        print(\"Background task completed!\")\n",
    "    return P(\"Task started\"), BackgroundTask(long_running_task)\n",
    "\n",
    "response = cli.get(\"/background\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "db0281cf",
   "metadata": {},
   "outputs": [],
   "source": [
    "test_eq(app.router.url_path_for('gday', nm='Jeremy'), '/user/Jeremy')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b827960a",
   "metadata": {},
   "outputs": [],
   "source": [
    "hxhdr = {'headers':{'hx-request':\"1\"}}\n",
    "\n",
    "@rt('/ft')\n",
    "def get(): return Title('Foo'),H1('bar')\n",
    "\n",
    "txt = cli.get('/ft').text\n",
    "assert '<title>Foo</title>' in txt and '<h1>bar</h1>' in txt and '<html>' in txt\n",
    "\n",
    "@rt('/xt2')\n",
    "def get(): return H1('bar')\n",
    "\n",
    "txt = cli.get('/xt2').text\n",
    "assert '<title>FastHTML page</title>' in txt and '<h1>bar</h1>' in txt and '<html>' in txt\n",
    "\n",
    "assert cli.get('/xt2', **hxhdr).text.strip() == '<h1>bar</h1>'\n",
    "\n",
    "@rt('/xt3')\n",
    "def get(): return Html(Head(Title('hi')), Body(P('there')))\n",
    "\n",
    "txt = cli.get('/xt3').text\n",
    "assert '<title>FastHTML page</title>' not in txt and '<title>hi</title>' in txt and '<p>there</p>' in txt"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "381a1b68",
   "metadata": {},
   "outputs": [],
   "source": [
    "@rt('/oops')\n",
    "def get(nope): return nope\n",
    "test_warns(lambda: cli.get('/oops?nope=1'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6b022adb",
   "metadata": {},
   "outputs": [],
   "source": [
    "def test_r(cli, path, exp, meth='get', hx=False, **kwargs):\n",
    "    if hx: kwargs['headers'] = {'hx-request':\"1\"}\n",
    "    test_eq(getattr(cli, meth)(path, **kwargs).text, exp)\n",
    "\n",
    "ModelName = str_enum('ModelName', \"alexnet\", \"resnet\", \"lenet\")\n",
    "fake_db = [{\"name\": \"Foo\"}, {\"name\": \"Bar\"}]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1b4b5717",
   "metadata": {},
   "outputs": [],
   "source": [
    "@rt('/html/{idx}')\n",
    "async def get(idx:int): return Body(H4(f'Next is {idx+1}.'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "dd017867",
   "metadata": {},
   "outputs": [],
   "source": [
    "@rt(\"/models/{nm}\")\n",
    "def get(nm:ModelName): return nm\n",
    "\n",
    "@rt(\"/files/{path}\")\n",
    "async def get(path: Path): return path.with_suffix('.txt')\n",
    "\n",
    "@rt(\"/items/\")\n",
    "def get(idx:int|None = 0): return fake_db[idx]\n",
    "\n",
    "@rt(\"/idxl/\")\n",
    "def get(idx:list[int]): return str(idx)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "52d9a3f2",
   "metadata": {},
   "outputs": [],
   "source": [
    "r = cli.get('/html/1', headers={'hx-request':\"1\"})\n",
    "assert '<h4>Next is 2.</h4>' in r.text\n",
    "test_r(cli, '/models/alexnet', 'alexnet')\n",
    "test_r(cli, '/files/foo', 'foo.txt')\n",
    "test_r(cli, '/items/?idx=1', '{\"name\":\"Bar\"}')\n",
    "test_r(cli, '/items/', '{\"name\":\"Foo\"}')\n",
    "assert cli.get('/items/?idx=g').text=='404 Not Found'\n",
    "assert cli.get('/items/?idx=g').status_code == 404\n",
    "test_r(cli, '/idxl/?idx=1&idx=2', '[1, 2]')\n",
    "assert cli.get('/idxl/?idx=1&idx=g').status_code == 404"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "bb8154cc",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'got sub/a.b'"
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "app = FastHTML()\n",
    "rt = app.route\n",
    "cli = TestClient(app)\n",
    "@app.route(r'/static/{path:path}.jpg')\n",
    "def index(path:str): return f'got {path}'\n",
    "@app.route(r'/static/{path:path}')\n",
    "def foo(path:str, a:int): return f'also got {path},{a}'\n",
    "cli.get('/static/sub/a.b.jpg').text"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "65a29643",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'also got sub/a.b,1'"
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "cli.get('/static/sub/a.b?a=1').text"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6bc9564c",
   "metadata": {},
   "outputs": [],
   "source": [
    "app.chk = 'foo'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b8be4ef3",
   "metadata": {},
   "outputs": [],
   "source": [
    "@app.get(\"/booly/\")\n",
    "def _(coming:bool=True): return 'Coming' if coming else 'Not coming'\n",
    "\n",
    "@app.get(\"/datie/\")\n",
    "def _(d:parsed_date): return d\n",
    "\n",
    "@app.get(\"/ua\")\n",
    "async def _(user_agent:str): return user_agent\n",
    "\n",
    "@app.get(\"/hxtest\")\n",
    "def _(htmx): return htmx.request\n",
    "\n",
    "@app.get(\"/hxtest2\")\n",
    "def _(foo:HtmxHeaders, req): return foo.request\n",
    "\n",
    "@app.get(\"/app\")\n",
    "def _(app): return app.chk\n",
    "\n",
    "@app.get(\"/app2\")\n",
    "def _(foo:FastHTML): return foo.chk,HttpHeader(\"mykey\", \"myval\")\n",
    "\n",
    "@app.get(\"/app3\")\n",
    "def _(foo:FastHTML): return HtmxResponseHeaders(location=\"http://example.org\")\n",
    "\n",
    "@app.get(\"/app4\")\n",
    "def _(foo:FastHTML): return Redirect(\"http://example.org\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5f045cf9",
   "metadata": {},
   "outputs": [],
   "source": [
    "test_r(cli, '/booly/?coming=true', 'Coming')\n",
    "test_r(cli, '/booly/?coming=no', 'Not coming')\n",
    "date_str = \"17th of May, 2024, 2p\"\n",
    "test_r(cli, f'/datie/?d={date_str}', '2024-05-17 14:00:00')\n",
    "test_r(cli, '/ua', 'FastHTML', headers={'User-Agent':'FastHTML'})\n",
    "test_r(cli, '/hxtest' , '1', headers={'HX-Request':'1'})\n",
    "test_r(cli, '/hxtest2', '1', headers={'HX-Request':'1'})\n",
    "test_r(cli, '/app' , 'foo')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "52f9f934",
   "metadata": {},
   "outputs": [],
   "source": [
    "r = cli.get('/app2', **hxhdr)\n",
    "test_eq(r.text, 'foo')\n",
    "test_eq(r.headers['mykey'], 'myval')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "26d316b6",
   "metadata": {},
   "outputs": [],
   "source": [
    "r = cli.get('/app3')\n",
    "test_eq(r.headers['HX-Location'], 'http://example.org')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "704aeba6",
   "metadata": {},
   "outputs": [],
   "source": [
    "r = cli.get('/app4', follow_redirects=False)\n",
    "test_eq(r.status_code, 303)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5fa127d5",
   "metadata": {},
   "outputs": [],
   "source": [
    "r = cli.get('/app4', headers={'HX-Request':'1'})\n",
    "test_eq(r.headers['HX-Redirect'], 'http://example.org')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "356db6c0",
   "metadata": {},
   "outputs": [],
   "source": [
    "@rt\n",
    "def meta():\n",
    "    return ((Title('hi'),H1('hi')),\n",
    "        (Meta(property='image'), Meta(property='site_name'))\n",
    "    )\n",
    "\n",
    "t = cli.post('/meta').text\n",
    "assert re.search(r'<body>\\s*<h1>hi</h1>\\s*</body>', t)\n",
    "assert '<meta' in t"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a3596991",
   "metadata": {},
   "outputs": [],
   "source": [
    "@app.post('/profile/me')\n",
    "def profile_update(username: str): return username\n",
    "\n",
    "test_r(cli, '/profile/me', 'Alexis', 'post', data={'username' : 'Alexis'})\n",
    "test_r(cli, '/profile/me', 'Missing required field: username', 'post', data={})"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "fdb9239c",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Example post request with parameter that has a default value\n",
    "@app.post('/pet/dog')\n",
    "def pet_dog(dogname: str = None): return dogname\n",
    "\n",
    "# Working post request with optional parameter\n",
    "test_r(cli, '/pet/dog', '', 'post', data={})"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3366d88f",
   "metadata": {},
   "outputs": [],
   "source": [
    "@dataclass\n",
    "class Bodie: a:int;b:str\n",
    "\n",
    "@rt(\"/bodie/{nm}\")\n",
    "def post(nm:str, data:Bodie):\n",
    "    res = asdict(data)\n",
    "    res['nm'] = nm\n",
    "    return res\n",
    "\n",
    "@app.post(\"/bodied/\")\n",
    "def bodied(data:dict): return data\n",
    "\n",
    "nt = namedtuple('Bodient', ['a','b'])\n",
    "\n",
    "@app.post(\"/bodient/\")\n",
    "def bodient(data:nt): return asdict(data)\n",
    "\n",
    "class BodieTD(TypedDict): a:int;b:str='foo'\n",
    "\n",
    "@app.post(\"/bodietd/\")\n",
    "def bodient(data:BodieTD): return data\n",
    "\n",
    "class Bodie2:\n",
    "    a:int|None; b:str\n",
    "    def __init__(self, a, b='foo'): store_attr()\n",
    "\n",
    "@rt(\"/bodie2/\", methods=['get','post'])\n",
    "def bodie(d:Bodie2): return f\"a: {d.a}; b: {d.b}\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "fcc024e6",
   "metadata": {},
   "outputs": [],
   "source": [
    "from fasthtml.xtend import Titled"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "48f0a45e",
   "metadata": {},
   "outputs": [],
   "source": [
    "d = dict(a=1, b='foo')\n",
    "\n",
    "test_r(cli, '/bodie/me', '{\"a\":1,\"b\":\"foo\",\"nm\":\"me\"}', 'post', data=dict(a=1, b='foo', nm='me'))\n",
    "test_r(cli, '/bodied/', '{\"a\":\"1\",\"b\":\"foo\"}', 'post', data=d)\n",
    "test_r(cli, '/bodie2/', 'a: 1; b: foo', 'post', data={'a':1})\n",
    "test_r(cli, '/bodie2/?a=1&b=foo&nm=me', 'a: 1; b: foo')\n",
    "test_r(cli, '/bodient/', '{\"a\":\"1\",\"b\":\"foo\"}', 'post', data=d)\n",
    "test_r(cli, '/bodietd/', '{\"a\":1,\"b\":\"foo\"}', 'post', data=d)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "73363a37",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Testing POST with Content-Type: application/json\n",
    "@app.post(\"/\")\n",
    "def index(it: Bodie): return Titled(\"It worked!\", P(f\"{it.a}, {it.b}\"))\n",
    "\n",
    "s = json.dumps({\"b\": \"Lorem\", \"a\": 15})\n",
    "response = cli.post('/', headers={\"Content-Type\": \"application/json\"}, data=s).text\n",
    "assert \"<title>It worked!</title>\" in response and \"<p>15, Lorem</p>\" in response"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0aeaac36",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Testing POST with Content-Type: application/json\n",
    "@app.post(\"/bodytext\")\n",
    "def index(body): return body\n",
    "\n",
    "response = cli.post('/bodytext', headers={\"Content-Type\": \"application/json\"}, data=s).text\n",
    "test_eq(response, '{\"b\": \"Lorem\", \"a\": 15}')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4034ee37",
   "metadata": {},
   "outputs": [],
   "source": [
    "files = [ ('files', ('file1.txt', b'content1')),\n",
    "         ('files', ('file2.txt', b'content2')) ]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "04881594",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "200\n",
      "content1,content2\n"
     ]
    }
   ],
   "source": [
    "@rt(\"/uploads\")\n",
    "async def post(files:list[UploadFile]):\n",
    "    return ','.join([(await file.read()).decode() for file in files])\n",
    "\n",
    "res = cli.post('/uploads', files=files)\n",
    "print(res.status_code)\n",
    "print(res.text)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "28a99667",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "200\n",
      "content1\n"
     ]
    }
   ],
   "source": [
    "res = cli.post('/uploads', files=[files[0]])\n",
    "print(res.status_code)\n",
    "print(res.text)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5bce33f0",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Set to 2025-11-19 10:04:20.016304\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "'Session time: 2025-11-19 10:04:20.016304'"
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "@rt(\"/setsess\")\n",
    "def get(sess, foo:str=''):\n",
    "    now = datetime.now()\n",
    "    sess['auth'] = str(now)\n",
    "    return f'Set to {now}'\n",
    "\n",
    "@rt(\"/getsess\")\n",
    "def get(sess): return f'Session time: {sess[\"auth\"]}'\n",
    "\n",
    "print(cli.get('/setsess').text)\n",
    "time.sleep(0.01)\n",
    "\n",
    "cli.get('/getsess').text"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "249fce31",
   "metadata": {},
   "outputs": [],
   "source": [
    "@rt(\"/sess-first\")\n",
    "def post(sess, name: str):\n",
    "    sess[\"name\"] = name\n",
    "    return str(sess)\n",
    "\n",
    "cli.post('/sess-first', data={'name': 2})\n",
    "\n",
    "@rt(\"/getsess-all\")\n",
    "def get(sess): return sess['name']\n",
    "\n",
    "test_eq(cli.get('/getsess-all').text, '2')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4d17ab9c",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "# Release notes\n"
     ]
    }
   ],
   "source": [
    "@rt(\"/upload\")\n",
    "async def post(uf:UploadFile): return (await uf.read()).decode()\n",
    "\n",
    "with open('../../CHANGELOG.md', 'rb') as f:\n",
    "    print(cli.post('/upload', files={'uf':f}, data={'msg':'Hello'}).text[:15])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "102c17ab",
   "metadata": {},
   "outputs": [],
   "source": [
    "@rt(\"/form-submit/{list_id}\")\n",
    "def options(list_id: str):\n",
    "    headers = {\n",
    "        'Access-Control-Allow-Origin': '*',\n",
    "        'Access-Control-Allow-Methods': 'POST',\n",
    "        'Access-Control-Allow-Headers': '*',\n",
    "    }\n",
    "    return Response(status_code=200, headers=headers)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "85bf2984",
   "metadata": {},
   "outputs": [],
   "source": [
    "h = cli.options('/form-submit/2').headers\n",
    "test_eq(h['Access-Control-Allow-Methods'], 'POST')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e4fc13b0",
   "metadata": {},
   "outputs": [],
   "source": [
    "from fasthtml.authmw import user_pwd_auth"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1f11eab1",
   "metadata": {},
   "outputs": [],
   "source": [
    "def _not_found(req, exc): return Div('nope')\n",
    "\n",
    "app,cli,rt = get_cli(FastHTML(exception_handlers={404:_not_found}))\n",
    "\n",
    "txt = cli.get('/').text\n",
    "assert '<div>nope</div>' in txt\n",
    "assert '<!doctype html>' in txt"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "063b7f43",
   "metadata": {},
   "outputs": [],
   "source": [
    "app,cli,rt = get_cli(FastHTML())\n",
    "\n",
    "@rt(\"/{name}/{age}\")\n",
    "def get(name: str, age: int):\n",
    "    return Titled(f\"Hello {name.title()}, age {age}\")\n",
    "\n",
    "assert '<title>Hello Uma, age 5</title>' in cli.get('/uma/5').text\n",
    "assert '404 Not Found' in cli.get('/uma/five').text"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "efd6fc6c",
   "metadata": {},
   "outputs": [],
   "source": [
    "auth = user_pwd_auth(testuser='spycraft')\n",
    "app,cli,rt = get_cli(FastHTML(middleware=[auth]))\n",
    "\n",
    "@rt(\"/locked\")\n",
    "def get(auth): return 'Hello, ' + auth\n",
    "\n",
    "test_eq(cli.get('/locked').text, 'not authenticated')\n",
    "test_eq(cli.get('/locked', auth=(\"testuser\",\"spycraft\")).text, 'Hello, testuser')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "32520197",
   "metadata": {},
   "outputs": [],
   "source": [
    "auth = user_pwd_auth(testuser='spycraft')\n",
    "app,cli,rt = get_cli(FastHTML(middleware=[auth]))\n",
    "\n",
    "@rt(\"/locked\")\n",
    "def get(auth): return 'Hello, ' + auth\n",
    "\n",
    "test_eq(cli.get('/locked').text, 'not authenticated')\n",
    "test_eq(cli.get('/locked', auth=(\"testuser\",\"spycraft\")).text, 'Hello, testuser')"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "967001f3",
   "metadata": {},
   "source": [
    "Check that afterware like toasts are called correctly."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "09a5fb96",
   "metadata": {},
   "outputs": [],
   "source": [
    "from fasthtml.toaster import *"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c7500555",
   "metadata": {},
   "outputs": [],
   "source": [
    "app,cli,rt = get_cli(FastHTML())\n",
    "\n",
    "# Modified toast functions to test resp logic\n",
    "def toast_after_(resp, req, sess):\n",
    "    if sk in sess and (not resp or isinstance(resp, (tuple,FT,FtResponse))):\n",
    "        sess['toast_duration'] = req.app.state.toast_duration\n",
    "        # We will activate the toast depending on the content of the response\n",
    "        if str('love') in str(resp): req.injects.append(render_toasts(sess))\n",
    "\n",
    "def setup_toasts_(app, duration=5000):\n",
    "    app.state.toast_duration = duration\n",
    "    app.hdrs += [Style(toast_css), Script(js)]\n",
    "    app.after.append(toast_after_)\n",
    "\n",
    "setup_toasts_(app)\n",
    "\n",
    "tm = 'Here is a toast!'\n",
    "\n",
    "@rt(\"/\")\n",
    "def index(sess): \n",
    "    add_toast(sess, tm, \"success\")\n",
    "    return P(\"I love toast!\")\n",
    "\n",
    "@rt(\"/no-toast\")\n",
    "def no_toast(sess):\n",
    "    add_toast(sess, tm, \"success\")\n",
    "    return P(\"I hate toast!\")\n",
    "\n",
    "r = cli.get('/no-toast')\n",
    "assert not tm in r.text\n",
    "\n",
    "r = cli.get('/')\n",
    "assert tm in r.text"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "bd0e4c06",
   "metadata": {},
   "source": [
    "Here's a quick sanity check to confirm that toasts are rendered in the toast container correctly. See [issue](https://github.com/AnswerDotAI/fasthtml/issues/762)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e2417e5f",
   "metadata": {},
   "outputs": [],
   "source": [
    "assert 'id=\"fh-toast-container\"' in r.text"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6a014add",
   "metadata": {},
   "source": [
    "## APIRouter"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d5223a9a",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "class RouteFuncs:\n",
    "    def __init__(self): super().__setattr__('_funcs', {})\n",
    "    def __setattr__(self, name, value): self._funcs[name] = value\n",
    "    def __getattr__(self, name):\n",
    "        if name in all_meths: raise AttributeError(\"Route functions with HTTP Names are not accessible here\")\n",
    "        try: return self._funcs[name]\n",
    "        except KeyError: raise AttributeError(f\"No route named {name} found in route functions\")\n",
    "    def __dir__(self): return list(self._funcs.keys())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0f08cb5e",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "class APIRouter:\n",
    "    \"Add routes to an app\"\n",
    "    def __init__(self, prefix:str|None=None, body_wrap=noop_body):\n",
    "        self.routes,self.wss = [],[]\n",
    "        self.rt_funcs = RouteFuncs()  # Store wrapped route function for discoverability\n",
    "        self.prefix = prefix if prefix else \"\"\n",
    "        self.body_wrap = body_wrap\n",
    "\n",
    "    def _wrap_func(self, func, path=None):\n",
    "        name = func.__name__\n",
    "        wrapped = _mk_locfunc(func, path)\n",
    "        wrapped.__routename__ = name\n",
    "        # If you are using the def get or def post method names, this approach is not supported\n",
    "        if name not in all_meths: setattr(self.rt_funcs, name, wrapped)\n",
    "        return wrapped\n",
    "\n",
    "    def __call__(self, path:str=None, methods=None, name=None, include_in_schema=True, body_wrap=None):\n",
    "        \"Add a route at `path`\"\n",
    "        def f(func):\n",
    "            p = self.prefix + (\"/\" + ('' if path.__name__=='index' else func.__name__) if callable(path) else path)\n",
    "            wrapped = self._wrap_func(func, p)\n",
    "            self.routes.append((func, p, methods, name, include_in_schema, body_wrap or self.body_wrap))\n",
    "            return wrapped\n",
    "        return f(path) if callable(path) else f\n",
    "\n",
    "    def __getattr__(self, name):\n",
    "        try: return getattr(self.rt_funcs, name)\n",
    "        except AttributeError: return super().__getattr__(self, name)\n",
    "\n",
    "    def to_app(self, app):\n",
    "        \"Add routes to `app`\"\n",
    "        for args in self.routes: app._add_route(*args)\n",
    "        for args in self.wss: app._add_ws(*args)\n",
    "\n",
    "    def ws(self, path:str, conn=None, disconn=None, name=None, middleware=None):\n",
    "        \"Add a websocket route at `path`\"\n",
    "        def f(func=noop): return self.wss.append((func, f\"{self.prefix}{path}\", conn, disconn, name, middleware))\n",
    "        return f"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "22e17ec8",
   "metadata": {},
   "outputs": [],
   "source": [
    "ar = APIRouter()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "61030ee4",
   "metadata": {},
   "outputs": [],
   "source": [
    "@ar(\"/hi\")\n",
    "def get(): return 'Hi there'\n",
    "@ar(\"/hi\")\n",
    "def post(): return 'Postal'\n",
    "@ar\n",
    "def ho(): return 'Ho ho'\n",
    "@ar(\"/hostie\")\n",
    "def show_host(req): return req.headers['host']\n",
    "@ar\n",
    "def yoyo(): return 'a yoyo'\n",
    "@ar\n",
    "def index(): return \"home page\"\n",
    "\n",
    "@ar.ws(\"/ws\")\n",
    "def ws(self, msg:str): return f\"Message text was: {msg}\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "cd413b0d",
   "metadata": {},
   "outputs": [],
   "source": [
    "app,cli,_ = get_cli(FastHTML())\n",
    "ar.to_app(app)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8c265ff8",
   "metadata": {},
   "outputs": [],
   "source": [
    "assert str(yoyo) == '/yoyo'\n",
    "# ensure route functions are properly discoverable on `APIRouter` and `APIRouter.rt_funcs`\n",
    "assert ar.prefix == ''\n",
    "assert str(ar.rt_funcs.index) == '/'\n",
    "assert str(ar.index) == '/'\n",
    "with ExceptionExpected(): ar.blah()\n",
    "with ExceptionExpected(): ar.rt_funcs.blah()\n",
    "# ensure any route functions named using an HTTPMethod are not discoverable via `rt_funcs`\n",
    "assert \"get\" not in ar.rt_funcs._funcs.keys()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8f7c5710",
   "metadata": {},
   "outputs": [],
   "source": [
    "test_eq(cli.get('/hi').text, 'Hi there')\n",
    "test_eq(cli.post('/hi').text, 'Postal')\n",
    "test_eq(cli.get('/hostie').text, 'testserver')\n",
    "test_eq(cli.post('/yoyo').text, 'a yoyo')\n",
    "\n",
    "test_eq(cli.get('/ho').text, 'Ho ho')\n",
    "test_eq(cli.post('/ho').text, 'Ho ho')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1236de78",
   "metadata": {},
   "outputs": [],
   "source": [
    "with cli.websocket_connect('/ws') as ws:\n",
    "    ws.send_text('{\"msg\":\"Hi!\"}')\n",
    "    data = ws.receive_text()\n",
    "    assert data == 'Message text was: Hi!'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "02a4e649",
   "metadata": {},
   "outputs": [],
   "source": [
    "ar2 = APIRouter(\"/products\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "151b9e3c",
   "metadata": {},
   "outputs": [],
   "source": [
    "@ar2(\"/hi\")\n",
    "def get(): return 'Hi there'\n",
    "@ar2(\"/hi\")\n",
    "def post(): return 'Postal'\n",
    "@ar2\n",
    "def ho(): return 'Ho ho'\n",
    "@ar2(\"/hostie\")\n",
    "def show_host(req): return req.headers['host']\n",
    "@ar2\n",
    "def yoyo(): return 'a yoyo'\n",
    "@ar2\n",
    "def index(): return \"home page\"\n",
    "\n",
    "@ar2.ws(\"/ws\")\n",
    "def ws(self, msg:str): return f\"Message text was: {msg}\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "77ce8548",
   "metadata": {},
   "outputs": [],
   "source": [
    "app,cli,_ = get_cli(FastHTML())\n",
    "ar2.to_app(app)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f1fc8425",
   "metadata": {},
   "outputs": [],
   "source": [
    "assert str(yoyo) == '/products/yoyo'\n",
    "assert ar2.prefix == '/products'\n",
    "assert str(ar2.rt_funcs.index) == '/products/'\n",
    "assert str(ar2.index) == '/products/'\n",
    "assert str(ar.index) == '/'\n",
    "with ExceptionExpected(): ar2.blah()\n",
    "with ExceptionExpected(): ar2.rt_funcs.blah()\n",
    "assert \"get\" not in ar2.rt_funcs._funcs.keys()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f265860d",
   "metadata": {},
   "outputs": [],
   "source": [
    "test_eq(cli.get('/products/hi').text, 'Hi there')\n",
    "test_eq(cli.post('/products/hi').text, 'Postal')\n",
    "test_eq(cli.get('/products/hostie').text, 'testserver')\n",
    "test_eq(cli.post('/products/yoyo').text, 'a yoyo')\n",
    "\n",
    "test_eq(cli.get('/products/ho').text, 'Ho ho')\n",
    "test_eq(cli.post('/products/ho').text, 'Ho ho')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e38d99cf",
   "metadata": {},
   "outputs": [],
   "source": [
    "with cli.websocket_connect('/products/ws') as ws:\n",
    "    ws.send_text('{\"msg\":\"Hi!\"}')\n",
    "    data = ws.receive_text()\n",
    "    assert data == 'Message text was: Hi!'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "259a0f53",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "for o in all_meths: setattr(APIRouter, o, partialmethod(APIRouter.__call__, methods=o))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f61d110c",
   "metadata": {},
   "outputs": [],
   "source": [
    "@ar.get\n",
    "def hi2(): return 'Hi there'\n",
    "@ar.get(\"/hi3\")\n",
    "def _(): return 'Hi there'\n",
    "@ar.post(\"/post2\")\n",
    "def _(): return 'Postal'\n",
    "\n",
    "@ar2.get\n",
    "def hi2(): return 'Hi there'\n",
    "@ar2.get(\"/hi3\")\n",
    "def _(): return 'Hi there'\n",
    "@ar2.post(\"/post2\")\n",
    "def _(): return 'Postal'"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "dfa6f859",
   "metadata": {},
   "source": [
    "## Extras"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7438435e",
   "metadata": {},
   "outputs": [],
   "source": [
    "app,cli,rt = get_cli(FastHTML(secret_key='soopersecret'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e518de83",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def cookie(key: str, value=\"\", max_age=None, expires=None, path=\"/\", domain=None, secure=False, httponly=False, samesite=\"lax\",):\n",
    "    \"Create a 'set-cookie' `HttpHeader`\"\n",
    "    cookie = cookies.SimpleCookie()\n",
    "    cookie[key] = value\n",
    "    if max_age is not None: cookie[key][\"max-age\"] = max_age\n",
    "    if expires is not None:\n",
    "        cookie[key][\"expires\"] = format_datetime(expires, usegmt=True) if isinstance(expires, datetime) else expires\n",
    "    if path is not None: cookie[key][\"path\"] = path\n",
    "    if domain is not None: cookie[key][\"domain\"] = domain\n",
    "    if secure: cookie[key][\"secure\"] = True\n",
    "    if httponly: cookie[key][\"httponly\"] = True\n",
    "    if samesite is not None:\n",
    "        assert samesite.lower() in [ \"strict\", \"lax\", \"none\", ], \"must be 'strict', 'lax' or 'none'\"\n",
    "        cookie[key][\"samesite\"] = samesite\n",
    "    cookie_val = cookie.output(header=\"\").strip()\n",
    "    return HttpHeader(\"set-cookie\", cookie_val)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "919e600a",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      " <!doctype html>\n",
      " <html>\n",
      "   <head>\n",
      "     <title>FastHTML page</title>\n",
      "     <link rel=\"canonical\" href=\"https://testserver/setcookie\">\n",
      "     <meta charset=\"utf-8\">\n",
      "     <meta name=\"viewport\" content=\"width=device-width, initial-scale=1, viewport-fit=cover\">\n",
      "<script src=\"https://cdn.jsdelivr.net/npm/htmx.org@2.0.7/dist/htmx.js\"></script><script src=\"https://cdn.jsdelivr.net/gh/answerdotai/fasthtml-js@1.0.12/fasthtml.js\"></script><script src=\"https://cdn.jsdelivr.net/gh/answerdotai/surreal@main/surreal.js\"></script><script src=\"https://cdn.jsdelivr.net/gh/gnat/css-scope-inline@main/script.js\"></script><script>\n",
      "    function sendmsg() {\n",
      "        window.parent.postMessage({height: document.documentElement.offsetHeight}, '*');\n",
      "    }\n",
      "    window.onload = function() {\n",
      "        sendmsg();\n",
      "        document.body.addEventListener('htmx:afterSettle',    sendmsg);\n",
      "        document.body.addEventListener('htmx:wsAfterMessage', sendmsg);\n",
      "    };</script>   </head>\n",
      "   <body></body>\n",
      " </html>\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "'Cookie was set at time 10:04:21.160325'"
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "@rt(\"/setcookie\")\n",
    "def get(req): return cookie('now', datetime.now())\n",
    "\n",
    "@rt(\"/getcookie\")\n",
    "def get(now:parsed_date): return f'Cookie was set at time {now.time()}'\n",
    "\n",
    "print(cli.get('/setcookie').text)\n",
    "time.sleep(0.01)\n",
    "cli.get('/getcookie').text"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8816f277",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def reg_re_param(m, s):\n",
    "    cls = get_class(f'{m}Conv', sup=StringConvertor, regex=s)\n",
    "    register_url_convertor(m, cls())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e9535978",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "# Starlette doesn't have the '?', so it chomps the whole remaining URL\n",
    "reg_re_param(\"path\", \".*?\")\n",
    "_static_exts = \"ico gif jpg jpeg webm css js woff png svg mp4 webp ttf otf eot woff2 txt html map pdf zip tgz gz csv mp3 wav ogg flac aac doc docx xls xlsx ppt pptx epub mobi bmp tiff avi mov wmv mkv xml yaml yml rar 7z tar bz2 htm xhtml apk dmg exe msi swf iso\".split()\n",
    "reg_re_param(\"static\", '|'.join(_static_exts))\n",
    "\n",
    "@patch\n",
    "def static_route_exts(self:FastHTML, prefix='/', static_path='.', exts='static'):\n",
    "    \"Add a static route at URL path `prefix` with files from `static_path` and `exts` defined by `reg_re_param()`\"\n",
    "    @self.route(f\"{prefix}{{fname:path}}.{{ext:{exts}}}\")\n",
    "    async def get(fname:str, ext:str): return FileResponse(f'{static_path}/{fname}.{ext}')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9a70842e",
   "metadata": {},
   "outputs": [],
   "source": [
    "reg_re_param(\"imgext\", \"ico|gif|jpg|jpeg|webm|pdf\")\n",
    "\n",
    "@rt(r'/static/{path:path}{fn}.{ext:imgext}')\n",
    "def get(fn:str, path:str, ext:str): return f\"Getting {fn}.{ext} from /{path}\"\n",
    "\n",
    "test_r(cli, '/static/foo/jph.me.ico', 'Getting jph.me.ico from /foo/')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8bb51383",
   "metadata": {},
   "outputs": [],
   "source": [
    "app.static_route_exts()\n",
    "assert 'These are the source notebooks for FastHTML' in cli.get('/README.txt').text"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b31de65a",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "@patch\n",
    "def static_route(self:FastHTML, ext='', prefix='/', static_path='.'):\n",
    "    \"Add a static route at URL path `prefix` with files from `static_path` and single `ext` (including the '.')\"\n",
    "    @self.route(f\"{prefix}{{fname:path}}{ext}\")\n",
    "    async def get(fname:str): return FileResponse(f'{static_path}/{fname}{ext}')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d4ff5919",
   "metadata": {},
   "outputs": [],
   "source": [
    "app.static_route('.md', static_path='../..')\n",
    "assert 'THIS FILE WAS AUTOGENERATED' in cli.get('/README.md').text"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f63b7a03",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "class StaticNoCache(StaticFiles):\n",
    "    def file_response(self, *args, **kwargs):\n",
    "        resp = super().file_response(*args, **kwargs)\n",
    "        resp.headers.setdefault(\"Cache-Control\", \"no-cache\")\n",
    "        return resp"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1960d7ff",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "class MiddlewareBase:\n",
    "    async def __call__(self, scope, receive, send) -> None:\n",
    "        if scope[\"type\"] not in [\"http\", \"websocket\"]:\n",
    "            await self._app(scope, receive, send)\n",
    "            return\n",
    "        return HTTPConnection(scope)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "83a20f93",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "class FtResponse:\n",
    "    \"Wrap an FT response with any Starlette `Response`\"\n",
    "    def __init__(self, content, status_code:int=200, headers=None, cls=HTMLResponse, media_type:str|None=None, background: BackgroundTask | None = None):\n",
    "        self.content,self.status_code,self.headers = content,status_code,headers\n",
    "        self.cls,self.media_type,self.background = cls,media_type,background\n",
    "\n",
    "    def __response__(self, req):\n",
    "        resp,kw = _part_resp(req, self.content)\n",
    "        cts = _xt_cts(req, resp)\n",
    "        tasks,httphdrs = kw.get('background'),kw.get('headers')\n",
    "        if not tasks: tasks = self.background\n",
    "        headers = {**(self.headers or {}), **httphdrs}\n",
    "        return self.cls(cts, status_code=self.status_code, headers=headers, media_type=self.media_type, background=tasks)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a9e66b7d",
   "metadata": {},
   "outputs": [],
   "source": [
    "@rt('/ftr')\n",
    "def get():\n",
    "    cts = Title('Foo'),H1('bar')\n",
    "    return FtResponse(cts, status_code=201, headers={'Location':'/foo/1'})\n",
    "\n",
    "r = cli.get('/ftr')\n",
    "\n",
    "test_eq(r.status_code, 201)\n",
    "test_eq(r.headers['location'], '/foo/1')\n",
    "txt = r.text\n",
    "assert '<title>Foo</title>' in txt and '<h1>bar</h1>' in txt and '<html>' in txt"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "94705ca7",
   "metadata": {},
   "source": [
    "Test on a single background task:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8cf7433e",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Starting slow task\n",
      "Finished slow task\n"
     ]
    }
   ],
   "source": [
    "def my_slow_task():\n",
    "    print('Starting slow task')    \n",
    "    time.sleep(0.001)\n",
    "    print('Finished slow task')        \n",
    "\n",
    "@rt('/background')\n",
    "def get():\n",
    "    return P('BG Task'), BackgroundTask(my_slow_task)\n",
    "\n",
    "r = cli.get('/background')\n",
    "\n",
    "test_eq(r.status_code, 200)\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "47c26feb",
   "metadata": {},
   "source": [
    "Test multiple background tasks:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "76d25483",
   "metadata": {},
   "outputs": [],
   "source": [
    "def increment(amount):\n",
    "    amount = amount/1000\n",
    "    print(f'Sleeping for {amount}s')    \n",
    "    time.sleep(amount)\n",
    "    print(f'Slept for {amount}s')     "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "bd7cf045",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Sleeping for 0.0s\n",
      "Slept for 0.0s\n",
      "Sleeping for 0.001s\n",
      "Slept for 0.001s\n",
      "Sleeping for 0.002s\n",
      "Slept for 0.002s\n"
     ]
    }
   ],
   "source": [
    "@rt\n",
    "def backgrounds():\n",
    "    tasks = BackgroundTasks()\n",
    "    for i in range(3): tasks.add_task(increment, i)\n",
    "    return P('BG Tasks'), tasks\n",
    "\n",
    "r = cli.get('/backgrounds')\n",
    "test_eq(r.status_code, 200)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5310fea9",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Sleeping for 0.0s\n",
      "Slept for 0.0s\n",
      "Sleeping for 0.001s\n",
      "Slept for 0.001s\n",
      "Sleeping for 0.002s\n",
      "Slept for 0.002s\n"
     ]
    }
   ],
   "source": [
    "@rt\n",
    "def backgrounds2():\n",
    "    tasks = [BackgroundTask(increment,i) for i in range(3)]\n",
    "    return P('BG Tasks'), *tasks\n",
    "\n",
    "r = cli.get('/backgrounds2')\n",
    "test_eq(r.status_code, 200)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "348f9bbd",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Sleeping for 0.0s\n",
      "Slept for 0.0s\n",
      "Sleeping for 0.001s\n",
      "Slept for 0.001s\n",
      "Sleeping for 0.002s\n",
      "Slept for 0.002s\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "{'status': 'done'}"
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "@rt\n",
    "def backgrounds3():\n",
    "    tasks = [BackgroundTask(increment,i) for i in range(3)]\n",
    "    return {'status':'done'}, *tasks\n",
    "\n",
    "r = cli.get('/backgrounds3')\n",
    "test_eq(r.status_code, 200)\n",
    "r.json()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9dc1025e",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def unqid(seeded=False):\n",
    "    id4 = UUID(int=random.getrandbits(128), version=4) if seeded else uuid4()\n",
    "    res = b64encode(id4.bytes)\n",
    "    return '_' + res.decode().rstrip('=').translate(str.maketrans('+/', '_-'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5b67e014",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def _add_ids(s):\n",
    "    if not isinstance(s, FT): return\n",
    "    if not getattr(s, 'id', None): s.id = unqid()\n",
    "    for c in s.children: _add_ids(c)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1f590f25",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "@patch\n",
    "def setup_ws(app:FastHTML, f=noop):\n",
    "    conns = {}\n",
    "    async def on_connect(scope, send): conns[scope.client] = send\n",
    "    async def on_disconnect(scope): conns.pop(scope.client)\n",
    "    app.ws('/ws', conn=on_connect, disconn=on_disconnect)(f)\n",
    "    async def send(s):\n",
    "        for o in conns.values(): await o(s)\n",
    "    app._send = send\n",
    "    return send"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a8a91edd",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "devtools_loc = \"/.well-known/appspecific/com.chrome.devtools.json\"\n",
    "\n",
    "@patch\n",
    "def devtools_json(self:FastHTML, path=None, uuid=None):\n",
    "    if not path: path = Path().absolute()\n",
    "    if not uuid: uuid = get_key()\n",
    "    @self.route(devtools_loc)\n",
    "    def devtools():\n",
    "        return dict(workspace=dict(root=path, uuid=uuid))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e27908c0",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "@patch\n",
    "def get_client(self:FastHTML, asink=False, **kw):\n",
    "    \"Get an httpx client with session cookes set from `**kw`\"\n",
    "    signer = itsdangerous.TimestampSigner(self.secret_key)\n",
    "    data = b64encode(dumps(kw).encode())\n",
    "    data = signer.sign(data)\n",
    "    client = httpx.AsyncClient() if asink else httpx.Client()\n",
    "    client.cookies.update({self.session_cookie: data.decode()})\n",
    "    return client"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "474e14b4",
   "metadata": {},
   "source": [
    "## Export -"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d211e8e2",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| hide\n",
    "import nbdev; nbdev.nbdev_export()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "bf47445c",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.12.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
